package com.ds.lens.flink.monitor.forecast;

import com.ds.lens.data.common.document.ForecastKafkaData;
import com.ds.lens.flink.monitor.common.CommonDeserializationSchema;
import com.ds.lens.flink.monitor.common.LensElasticsearchSink;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.forecast.function.ForecastDataSinkFunction;
import com.ds.lens.flink.monitor.forecast.function.ForecastKafkaDataFunction;
import com.ds.lens.flink.monitor.forecast.function.ForecastKafkaDataSinkFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.http.HttpHost;

import java.util.List;
import java.util.Properties;

/**
 * Description:
 *
 * @author WeiShaoying
 * @date 2020-05-08
 */
public class ForecastDataIntakeJob {

    public static void main(String[] args) throws Exception {
        ForecastConfig forecastConfig = MonitorJobContext.getInstance().getForecastConfig();
        Properties forecastProperties = forecastConfig.getKafkaProperties();
        if (forecastProperties == null) {
            System.err.println("No forecast kafka properties found");
            return;
        }

        List<HttpHost> transportAddresses = MonitorJobContext.getInstance().getEsTransportAddresses();
        String forecastDataTopic = forecastConfig.getForecastDataTopic();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(MonitorJobContext.getInstance().getMetricsJobParallelism());
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        DataStream<ForecastKafkaData> dataStream = env.addSource(new FlinkKafkaConsumer<>(forecastDataTopic,
                new CommonDeserializationSchema<>(ForecastKafkaData.class), forecastProperties))
                .process(new ForecastKafkaDataFunction());
        dataStream.addSink(new LensElasticsearchSink<>(transportAddresses, new ForecastKafkaDataSinkFunction(), 100, 10000L).build());
        dataStream.addSink(new LensElasticsearchSink<>(transportAddresses, new ForecastDataSinkFunction(), 100, 10000L).build());

        env.execute(ForecastDataIntakeJob.class.getSimpleName());
    }
}
